package edu.npu.GraphTransform;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;

public class GraphMetisQualify {
	public static final String SEPARATOR = "\t";
	public static final String FLAG = "#";
	//将拓扑的key改为ID形式
	public static class Key_MetisQualifyMapper extends 
					Mapper<LongWritable, Text, LongWritable, Text> {
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			long k;
			String line = value.toString();
			String[] splits = line.split(SEPARATOR);
			if(splits[0].startsWith("#")) {
				k = Long.parseLong(splits[1]);
				context.write(new LongWritable(k), value);
				context.getCounter(COUNTER.NODE).increment(1);
			} else {
				k = Long.parseLong(splits[0]);
				context.write(new LongWritable(k), new Text(splits[1]));
				context.getCounter(COUNTER.EDGE).increment(1);
			}
		}		
	}
	public static class Key_MetisQualifyReducer extends 
							Reducer<LongWritable, Text, LongWritable, Text> {
		@Override
		protected void reduce(LongWritable key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			String qualifyKey = null;
			List<Text> list = new ArrayList<Text>();//用来存储遍历过的节点
			for(Text value : values) {
				if(value.toString().startsWith("#")) {
					qualifyKey = value.toString().split(SEPARATOR)[0].substring(1);
					break;
				} else
					list.add(new Text(value));
			}
			for(Text value : values) 
					context.write(new LongWritable(Long.parseLong(qualifyKey)), 
								value);
			for(Text value : list) 
				context.write(new LongWritable(Long.parseLong(qualifyKey)), 
						value);
		}
	}
	//将拓扑的value改为ID形式
	public static class Value_MetisQualifyMapper extends
			Mapper<LongWritable, Text, LongWritable, Text> {
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			long v;
			String line = value.toString();
			String[] splits = line.split(SEPARATOR);
			if (splits[0].startsWith("#")) {
				v = Long.parseLong(splits[1]);
				context.write(new LongWritable(v), value);
			} else {
				v = Long.parseLong(splits[1]);
				context.write(new LongWritable(v), new Text(splits[0]));
			}
		}
	}
	public static class Value_MetisQualifyReducer extends
			Reducer<LongWritable, Text, LongWritable, Text> {
		@Override
		protected void reduce(LongWritable key, Iterable<Text> values,
				Context context) throws IOException, InterruptedException {
			String qualifyValue = null;
			List<Text> list = new ArrayList<Text>();
			for (Text value : values) {
				if (value.toString().startsWith("#")) {
					qualifyValue = value.toString().split(SEPARATOR)[0].substring(1);
					break;
				} else
					list.add(new Text(value));
			}
			for (Text value : values) {
				context.write(new LongWritable(Long.parseLong(value.toString())),
						new Text(qualifyValue));
			}
			for(Text value : list)
				context.write(new LongWritable(Long.parseLong(value.toString())),
						new Text(qualifyValue));
		}
	}
	
	//数据变成metis所需的形式
	public static class MetisQualifyMapper extends Mapper<LongWritable, Text, LongWritable, LongWritable> {
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] splits = line.split(SEPARATOR);
			context.write(new LongWritable(Long.parseLong(splits[0])), 
					new LongWritable(Long.parseLong(splits[1])));
			context.write(new LongWritable(Long.parseLong(splits[1])), 
					new LongWritable(Long.parseLong(splits[0])));
		}
	}
	public static class MetisQualifyPartitioner extends Partitioner<LongWritable, LongWritable> {
		@Override
		public int getPartition(LongWritable key, LongWritable value, int num) {
			int MAXIMUM = 41652231;
			return (int) (key.get() * num * 1.0 / MAXIMUM);
		}
	}
	public static class MetisQualifyReducer extends Reducer<LongWritable, LongWritable, LongWritable, Text> {
		@Override
		protected void reduce(LongWritable key, Iterable<LongWritable> values, Context context)
				throws IOException, InterruptedException {
			StringBuilder sb = new StringBuilder();
			for(LongWritable value : values) 
				sb.append(value + " ");
			context.write(key, new Text(sb.substring(0, sb.length() - 1)));
		}
	}
	//用于counter的类
	public static enum COUNTER {
		EDGE,
		NODE
	};
	//用于去重的类
	public static class Pair implements WritableComparable<Pair> {
		public LongWritable v1, v2;
		@Override
		public int compareTo(Pair other) {
			if(this.v1.compareTo(other.v1) < 0)
				return -1;
			if(this.v1.compareTo(other.v1) > 0)
				return 1;
			if(this.v2.compareTo(other.v2) < 0)
				return -1;
			if(this.v2.compareTo(other.v2) > 0)
				return 1;
			return 0;
		}
		public Pair() {
			this.v1 = new LongWritable();
			this.v2 = new LongWritable();
		}
		public Pair(LongWritable v1, LongWritable v2) {
			this.v1 = v1;
			this.v2 = v2;
		}
		@Override
		public void readFields(DataInput in) throws IOException {
			v1.readFields(in);
			v2.readFields(in);
		}
		@Override
		public void write(DataOutput out) throws IOException {
			v1.write(out);
			v2.write(out);
		}
		@Override
		public String toString() {
			return v1.get() + "\t" + v2.get();
		}
		
	}
}
